package com.flc.archive.hdfs;

import java.io.IOException;

import java.net.URI;
import java.net.URISyntaxException;

import com.flc.archive.config.AppConfig;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 把本地文件（如windows或linux文件拷贝到hdfs上）
 * @author haizi
 *
 */
public class CopyLocalFileToHdfs implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(CopyLocalFileToHdfs.class);

    private String source;
    private String dest;

    public CopyLocalFileToHdfs(String source, String dest) {
        this.source = source;
        this.dest = dest;
    }

    /**
     * @function 本地文件上传至 HDFS
     * @param source 原文件路径
     * @param dest  目的文件路径
     * @throws IOException
     * @throws URISyntaxException
     */
    private void copyFromLocal(String source, String dest) throws URISyntaxException, IOException {
        AppConfig.HdfsCfg hdfsCfg = AppConfig.getInstance().getHdfsCfg();
        // 读取hadoop文件系统的配置
        Configuration conf = new Configuration();
        URI uri = new URI(hdfsCfg.getHdfsAddress());
        // FileSystem是用户操作HDFS的核心类，它获得URI对应的HDFS文件系统
        FileSystem fileSystem = FileSystem.get(uri, conf);
        // 源文件路径
        Path srcPath = new Path(source);
        // 目的路径
        Path dstPath = new Path(dest);
        // 查看目的路径是否存在
        if (!(fileSystem.exists(dstPath))) {
            // 如果路径不存在，即刻创建
            fileSystem.mkdirs(dstPath);
        }
        // 得到本地文件名称
        String filename = source.substring(source.lastIndexOf('/') + 1, source.length());
        try {
            // 将本地文件上传到HDFS
            fileSystem.copyFromLocalFile(srcPath, dstPath);
            logger.info("File " + filename + " copied to " + dest);
        } catch (Exception e) {
            e.printStackTrace();
            System.exit(1);
        } finally {
            fileSystem.close();
        }
    }

    @Override
    public void run() {
        try {
            copyFromLocal(source, dest);
        } catch (URISyntaxException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}